package cn.com.cennavi.service;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Set;

import org.apache.spark.api.java.Optional;
import org.apache.spark.api.java.function.Function3;
import org.apache.spark.streaming.State;

import com.vividsolutions.jts.geom.Coordinate;
import com.vividsolutions.jts.geom.Geometry;
import com.vividsolutions.jts.geom.Point;

import cn.com.cennavi.bean.DiDiNewData1;
import cn.com.cennavi.bean.DidiQueue2;
import cn.com.cennavi.bean.DidiQueueStatus2;
import cn.com.cennavi.bean.DifPointTotalMapInfo;
import cn.com.cennavi.util.MapUtil;
import scala.Tuple2;

public class StateUpdate1 implements Function3<String, Optional<Tuple2<DiDiNewData1, Iterable<DifPointTotalMapInfo>>>, State<DidiQueueStatus2>, Tuple2<String, DidiQueue2>> {
	private static final long serialVersionUID = 1L;

	@Override
	public Tuple2<String, DidiQueue2> call(String carID, Optional<Tuple2<DiDiNewData1, Iterable<DifPointTotalMapInfo>>> arg1, State<DidiQueueStatus2> status) throws Exception {
		// TODO Auto-generated method stub
		DidiQueue2 returnQueue = null;
		DidiQueueStatus2 returnStatus = null;
		if (status.exists()) {
			returnStatus = status.get();
		}

		if (returnStatus == null) {
			returnStatus = new DidiQueueStatus2();
		}
		if (arg1.isPresent()) {
			Tuple2<DiDiNewData1, Iterable<DifPointTotalMapInfo>> value = arg1.get();
			// 属于北京数据
			// 缓存进入缓冲区的点序列和备选道路形状
			returnQueue = putEnterBufDiDi(value._1, returnStatus, value._2);
		}
		status.update(returnStatus);
		if (returnQueue != null) {
			return new Tuple2<String, DidiQueue2>(carID, returnQueue);
		} else {
			return null;
		}
	}

	private DidiQueue2 putEnterBufDiDi(DiDiNewData1 didi, DidiQueueStatus2 returnStatus, Iterable<DifPointTotalMapInfo> mapInfo) {
		// TODO Auto-generated method stub
		DidiQueue2 returnQueue = null;
		/** 如果gps点落在某一条或多条道路的缓冲区域内，写入缓存序列 **/
		/** 判断是不是在该网格中分歧点道路缓冲区内，key：pointId+dir **/
		HashMap<String, DifPointTotalMapInfo> linkGeoList = judgeIsInBufferArea(didi, mapInfo,returnStatus);
		didi.difPointId = getDifPointId(linkGeoList);
		/**
		 * 只有匹配完后才会有cachegps点 如果cache缓冲区有缓存的gps点，考虑是不是应该加入到enterBufQueue中
		 * 1：第一次进来后，如果gps点有缓冲区，则判断是不是需要添加cache。如果没有缓冲区则直接pass。
		 **/
		if (returnStatus.cacheQueue.size() != 0) {
			if (!didi.difPointId.equals("")) {
				judgeIsInCacheBufferArea(didi, returnStatus);
				returnStatus.enterBufQueue.add(didi);
				/** 讲该辆车的所属缓冲区保存 **/
				returnStatus.difPoint_dir_bufferMap = linkGeoList;
			}
			returnStatus.cacheQueue.clear();
		} else {
			// 之前没有缓存点了，第二个点进来
			HashMap<String, DifPointTotalMapInfo> linkGeoList_bef = returnStatus.difPoint_dir_bufferMap;
			if (returnStatus.enterBufQueue.size() == 0 && didi.difPointId.equals("")) {
				return null;
			}
			if (returnStatus.enterBufQueue.size() == 0 && !didi.difPointId.equals("")) {
				returnStatus.enterBufQueue.add(didi);
				/** 讲该辆车的所属缓冲区保存 **/
				returnStatus.difPoint_dir_bufferMap = linkGeoList;
				return null;
			}
			if (returnStatus.enterBufQueue.size() != 0 && didi.difPointId.equals("")) {
				if (returnStatus.car_keepErrorPoint >= 2) {
					// 生成输出点序对象
					returnQueue = new DidiQueue2();
					returnStatus.enterBufQueue.add(didi);
					returnQueue.enterBufQueue.addAll(returnStatus.enterBufQueue);

					returnQueue.cacheQueue.addAll(returnStatus.cacheQueue);
					returnQueue.difPoint_dir_bufferMap.putAll(returnStatus.difPoint_dir_bufferMap);
					returnStatus.car_keepErrorPoint = 0;
					// 保留最后一个点的缓冲区
					returnStatus.difPoint_dir_bufferMap.clear();
					returnStatus.difPoint_dir_bufferMap.putAll(linkGeoList);
					// System.out.println("niuxinzan:保存这次最后一个点的分歧点id:" +
					// linkGeoList.keySet().toString());
					// 保留这次匹配的最后10个点
					updateCacheDiDi(returnStatus.enterBufQueue, returnStatus);
					// 清空enterbuffer点
					returnStatus.enterBufQueue.clear();
				} else {
					returnStatus.enterBufQueue.add(didi);
					returnStatus.car_keepErrorPoint++;
					return null;
				}
			}
			if (returnStatus.enterBufQueue.size() != 0 && !didi.difPointId.equals("")) {
				if (judgeBuffer(linkGeoList_bef, linkGeoList, didi)) {
					returnStatus.enterBufQueue.add(didi);
					returnStatus.difPoint_dir_bufferMap = linkGeoList_bef;
					returnStatus.car_keepErrorPoint = 0;
				} else {
					if (returnStatus.car_keepErrorPoint >= 2) {
						// 生成输出点序对象
						returnQueue = new DidiQueue2();
						returnStatus.enterBufQueue.add(didi);
						returnQueue.enterBufQueue.addAll(returnStatus.enterBufQueue);

						returnQueue.cacheQueue.addAll(returnStatus.cacheQueue);
						returnQueue.difPoint_dir_bufferMap.putAll(returnStatus.difPoint_dir_bufferMap);
						returnStatus.car_keepErrorPoint = 0;
						// System.out.println("niuxinzan:kaishi pipei:");

						// 保留最后一个点的缓冲区
						returnStatus.difPoint_dir_bufferMap.clear();
						returnStatus.difPoint_dir_bufferMap.putAll(linkGeoList);
						// System.out.println("niuxinzan:保存这次最后一个点的分歧点id:" +
						// linkGeoList.keySet().toString());
						// 保留这次匹配的最后10个点
						updateCacheDiDi(returnStatus.enterBufQueue, returnStatus);
						// 清空enterbuffer点
						returnStatus.enterBufQueue.clear();
					} else {
						returnStatus.enterBufQueue.add(didi);
						returnStatus.car_keepErrorPoint++;
					}
				}
			}
		}
		return returnQueue;
	}

	private void judgeIsInCacheBufferArea(DiDiNewData1 didi, DidiQueueStatus2 returnStatus) {
		// TODO Auto-generated method stub
		// boolean isTrue = false;
		/**
		 * 如果cache缓存点最后一个点与didi的点属于同一个缓冲区则将cache数据加入到enterBufQueue，否则舍弃
		 */
		LinkedList<DiDiNewData1> cacheQueue = returnStatus.cacheQueue;
		int size = cacheQueue.size();
		String cacheDifPointId = cacheQueue.get(size - 1).difPointId;
		String didiDifPointId = didi.difPointId;
		if (!didiDifPointId.equals("") && !cacheDifPointId.equals("")) {
			String[] difIdArr = cacheDifPointId.split("\\|");
			for (String id : difIdArr) {
				didiDifPointId.contains(id);
				returnStatus.enterBufQueue.addAll(returnStatus.cacheQueue);
				break;
			}
		}
		returnStatus.cacheQueue.clear();
	}

	private void removeKey(String cacheDifPointId, HashMap<String, DifPointTotalMapInfo> difPoint_dir_bufferMap) {
		// TODO Auto-generated method stub
		Set<String> set = difPoint_dir_bufferMap.keySet();
		Set<String> set1 = new HashSet<String>();
		for (String id : set) {
			if (!cacheDifPointId.contains(id)) {
				set1.add(id);
			}
		}
		for (String sss : set1) {
			difPoint_dir_bufferMap.remove(sss);
		}
	}

	/**
	 * 距离现在最近的点往前推，寻找属于公共分叉口的nodeid
	 * 
	 * @param cacheQueue
	 * @return
	 */
	private String getCacheDifPointId(LinkedList<DiDiNewData1> cacheQueue) {
		// TODO Auto-generated method stub
		int size = cacheQueue.size();
		String cacheDifPointId = cacheQueue.get(size - 1).difPointId;
		if (cacheDifPointId == null) {
			return null;
		}
		for (int i = size - 1; i > 0; i--) {
			String difPointIds = cacheQueue.get(i).difPointId;
			if (difPointIds == null) {
				continue;
			}
			/**
			 * 判断前一个点与后一个点是否有共同的buffer
			 */
			boolean isContain = judgeIsContain(cacheDifPointId, difPointIds);
			if (isContain) {
				cacheDifPointId = getCacheDifPointId(cacheDifPointId, difPointIds);
			} else {
				return cacheDifPointId;
			}
		}
		return cacheDifPointId;
	}

	private String getCacheDifPointId(String cacheDifPointId, String difPointIds) {
		// TODO Auto-generated method stub
		StringBuffer sb = new StringBuffer();
		String[] difPointIdsArr = cacheDifPointId.split("\\|");
		for (String id : difPointIdsArr) {
			if (difPointIds.contains(id)) {
				sb.append(id);
				sb.append("|");
			}
		}
		return sb.toString();
	}

	private boolean judgeIsContain(String difPointIds, String difPointIds_bef) {
		// TODO Auto-generated method stub
		String[] difPointIdsArr = difPointIds.split("\\|");
		if (difPointIds == null || difPointIds.equals("")) {
			return false;
		}
		for (String id : difPointIdsArr) {
			if (difPointIds_bef.contains(id)) {
				return true;
			}
		}
		return false;
	}

	private String getDifPointId(HashMap<String, DifPointTotalMapInfo> linkGeoList) {
		// TODO Auto-generated method stub
		if (linkGeoList.size() == 0) {
			return "";
		}
		StringBuffer sf = new StringBuffer();
		Set<String> difpointId_dir = linkGeoList.keySet();
		for (String difpoint : difpointId_dir) {
			sf.append(difpoint);
			sf.append("|");
		}
		String difPointId = sf.toString();
		if (difPointId.endsWith("|")) {
			difPointId = difPointId.substring(0, difPointId.length() - 1).toString();
		}
		return difPointId;
	}

	/**
	 * 判断是否属于相同的缓冲区
	 * 
	 * @param didi
	 */
	private boolean judgeBuffer(HashMap<String, DifPointTotalMapInfo> linkGeoList_bef, HashMap<String, DifPointTotalMapInfo> linkGeoList_now, DiDiNewData1 didi) {
		// TODO Auto-generated method stub
		// System.out.println("2.2.1-----开始---当前后缓冲区有交集" + ":" + new
		// Date().getTime());
		if (linkGeoList_bef == null || linkGeoList_bef.size() == 0) {
			System.out.println("这个地方不应该走！");
			return false;
		}

		ArrayList<String> difKeyList = new ArrayList<>();
		HashMap<String, DifPointTotalMapInfo> linkGeoMap = new HashMap<>();
		StringBuffer sb = new StringBuffer();
		for (Map.Entry<String, DifPointTotalMapInfo> entry : linkGeoList_bef.entrySet()) {
			String key = entry.getKey();
			// System.out.println("key1:" + key);
			if (linkGeoList_now.containsKey(key)) {
				linkGeoMap.put(key, entry.getValue());
				sb.append(key);
				sb.append("|");
			} else {
				difKeyList.add(key);
			}
		}
		// String difPointId = sb.toString();
		// if (difPointId.endsWith("|")) {
		// difPointId = difPointId.substring(0, difPointId.length() - 1);
		// }
		// didi.difPointId = difPointId;
		if (linkGeoMap.size() != 0) {
			for (String key : difKeyList) {
				linkGeoList_bef.remove(key);
			}
			return true;

		}
		return false;
	}

	private HashMap<String, DifPointTotalMapInfo> judgeIsInBufferArea(DiDiNewData1 didi, Iterable<DifPointTotalMapInfo> mapInfo, DidiQueueStatus2 returnStatus) {
		// TODO Auto-generated method stub
		LinkedList<DiDiNewData1> queue=returnStatus.enterBufQueue;
		int num=queue.size();
		DiDiNewData1 beforeDidi=null;
		int count=100;
		if(num!=0){
			 beforeDidi=queue.get(num-1);
		}else{
			count=0;
		}
		Point point = MapUtil.getPoint(new Coordinate(didi.lon, didi.lat));
		HashMap<String, DifPointTotalMapInfo> dataList = new HashMap<>();
		for (DifPointTotalMapInfo entry : mapInfo) {
			Geometry geo = entry.roadGeo;		
			if (geo.contains(point)) {
				if(count==0){
					dataList.put(entry.dif_Point_Id + "_" + entry.flag, entry);
				}else{
					double direction = MapUtil.azimuth(beforeDidi.lon, beforeDidi.lat, didi.lon, didi.lat);
					double enter = entry.enter_Azimuth;
					double outer = entry.out_Azimuth;
					String minMaxAngle11 = getminMaxAngle(enter);
					String minMaxAngle22 = getminMaxAngle(outer);
					if (judgeAngle(minMaxAngle11, minMaxAngle22, direction)) {
						dataList.put(entry.dif_Point_Id + "_" + entry.flag, entry);
					} else {
						System.out.println("舍弃");
					}
				}
				
			}
		}

		return dataList;
	}

	private boolean judgeAngle(String minMaxAngle1, String minMaxAngle2, double direction) {
		// TODO Auto-generated method stub
		// TODO Auto-generated method stub
				String[] s1 = minMaxAngle1.split("\\:");
				String[] s2 = minMaxAngle2.split("\\:");
				for (String ss : s1) {
					String[] sss = ss.split("\\_");
					double minAngle = Double.parseDouble(sss[0]);
					double maxAngle = Double.parseDouble(sss[1]);
					if (direction >= minAngle && direction <= maxAngle) {
						return true;
					}
				}
				for (String ss : s2) {
					String[] sss = ss.split("\\_");
					double minAngle = Double.parseDouble(sss[0]);
					double maxAngle = Double.parseDouble(sss[1]);
					if (direction >= minAngle && direction <= maxAngle) {
						return true;
					}
				}
				return false;
	}

	private String getminMaxAngle(double enter) {
		// TODO Auto-generated method stub
		// 角度在第一象限
				if (enter >= 0 && enter <= 90) {
					double s1 = enter - 90;
					double s2 = enter + 90;
					return s1 + "_" + s2;
				}

				// 第二象限
				if (enter > 90 && enter <= 180) {
					double s1 = enter - 270;
					double s2 = enter - 90;
					return "-180_" + s1 + ":" + s2 + "_" + 180;
				}
				// 第三象限
				if (enter >= -180 && enter < -90) {
					double s1 = enter + 90;
					double s2 = enter +270;
					return "-180_" + s1 + ":" + s2 + "_" + 180;
				}

				// 第四象限
				if (enter >= -90 && enter < 0) {
					double s1 = enter - 90;
					double s2 = enter + 90;
					return s1 + "_" + s2;
				}
				return null;
	}

	/**
	 * 只保留上次匹配成功的最后10个gps点
	 * 
	 * @param enterBufQueue
	 * @param returnStatus
	 */
	private void updateCacheDiDi(LinkedList<DiDiNewData1> enterBufQueue, DidiQueueStatus2 returnStatus) {
		// TODO Auto-generated method stub
		int size = enterBufQueue.size();
		List<DiDiNewData1> list1 = enterBufQueue.subList((size - 10) > 0 ? (size - 10) : 0, size);
		returnStatus.cacheQueue.clear();
		for (DiDiNewData1 object : list1) {
			returnStatus.cacheQueue.add(object);
		}
	}

	public static void main(String[] args) {

	}
}
